package org.apache.rocketmq.remoting.protocol;

import com.alibaba.fastjson.annotation.JSONField;
import com.google.common.base.Stopwatch;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.annotation.CFNotNull;
import org.apache.rocketmq.remoting.common.BoundaryType;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;

import java.lang.annotation.Annotation;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Modifier;
import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;



/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2024/12/27
 * @方法描述：Remoting协议中封装网络数据的对象，请求体和响应体都是由这个类的对象来表示的
 */
public class RemotingCommand {

    public static final String SERIALIZE_TYPE_PROPERTY = "rocketmq.serialize.type";

    public static final String SERIALIZE_TYPE_ENV = "ROCKETMQ_SERIALIZE_TYPE";

    public static final String REMOTING_VERSION_KEY = "rocketmq.remoting.version";

    static final Logger log = LoggerFactory.getLogger(LoggerName.ROCKETMQ_REMOTING_NAME);

    //用来计算请求类型的成员变量，因为RemotingCommand即可表示请求也可以表示响应
    //在处理RemotingCommand消息的时候要先判断接收到的RemotingCommand是请求还是响应
    //而这个成员变量就是用来设置当前RemotingCommand是请求还是响应
    private static final int RPC_TYPE = 0;

    //用来计算当前RPC操作是否为单向RPC操作
    private static final int RPC_ONEWAY = 1;

    //缓存请求类型与该类型中所有字段映射关系的map
    //比如说GetRouteInfoRequestHeader类以及它的父类中总共有七个属性，也就是7个字段
    //那么这个map中就会存放key为GetRouteInfoRequestHeader类型，value为存放了7个对应字段的Field数组
    private static final Map<Class<? extends CommandCustomHeader>, Field[]> CLASS_HASH_MAP = new HashMap<>();

    //以下这几个成员变量会在解码请求头的时候用到，大家可以从当前类的decodeCommandCustomHeader()开始看起，看看以下这些成员变量是如何使用的
    //我就不单独给这几个成员变量添加注释了，从具体的方法中大家很容易就能了解这几个成员变量的作用


    private static final Map<Class, String> CANONICAL_NAME_CACHE = new HashMap<>();

    private static final Map<Field, Boolean> NULLABLE_FIELD_CACHE = new HashMap<>();

    private static final String STRING_CANONICAL_NAME = String.class.getCanonicalName();

    private static final String DOUBLE_CANONICAL_NAME_1 = Double.class.getCanonicalName();

    private static final String DOUBLE_CANONICAL_NAME_2 = double.class.getCanonicalName();

    private static final String INTEGER_CANONICAL_NAME_1 = Integer.class.getCanonicalName();

    private static final String INTEGER_CANONICAL_NAME_2 = int.class.getCanonicalName();

    private static final String LONG_CANONICAL_NAME_1 = Long.class.getCanonicalName();

    private static final String LONG_CANONICAL_NAME_2 = long.class.getCanonicalName();

    private static final String BOOLEAN_CANONICAL_NAME_1 = Boolean.class.getCanonicalName();

    private static final String BOOLEAN_CANONICAL_NAME_2 = boolean.class.getCanonicalName();

    private static final String BOUNDARY_TYPE_CANONICAL_NAME = BoundaryType.class.getCanonicalName();

    //配置版本号
    private static volatile int configVersion = -1;

    //生成唯一请求Id的成员变量
    private static AtomicInteger requestId = new AtomicInteger(0);

    //消息体的序列化方式
    private static SerializeType serializeTypeConfigInThisServer = SerializeType.JSON;

    //从系统环境变量中获得用户配置的消息序列化方式
    static {
        final String protocol = System.getProperty(SERIALIZE_TYPE_PROPERTY, System.getenv(SERIALIZE_TYPE_ENV));
        //如果用户配置了序列化方式，那么更新serializeTypeConfigInThisServer成员变量的值
        if (!StringUtils.isBlank(protocol)) {
            try {
                serializeTypeConfigInThisServer = SerializeType.valueOf(protocol);
            } catch (IllegalArgumentException e) {
                throw new RuntimeException("parser specified protocol error. protocol=" + protocol, e);
            }
        }
    }

    //当前消息如果是请求，那么这个成员变量就是请求码，代表请求的请求类型
    private int code;

    //定义使用的编程语言类型
    private LanguageCode language = LanguageCode.JAVA;

    //当前消息的版本号，其实就代表当前协议的版本号
    private int version = 0;

    //请求唯一标识，这里可以看出来每创建一个RemotingCommand对象的时候
    //请求的唯一Id就已经被赋值了，并且请求Id是唯一递增的
    private int opaque = requestId.getAndIncrement();

    //这个标志位就是用来计算当前消息是请求消息还是响应消息的
    private int flag = 0;

    //定义消息的备注信息
    private String remark;

    //存储消息额外信息的map
    private HashMap<String, String> extFields;

    //请求消息的请求头，在发送请求的时候，请求头会赋值给这个成员变量
    //如果是响应对象，那么响应头就会赋值给这个成员变量
    private transient CommandCustomHeader customHeader;

    //在处理请求的时候，要对消息解码，解码后得到的请求头就会赋值给这个成员变量
    private transient CommandCustomHeader cachedHeader;

    //当前消息序列化方式
    private SerializeType serializeTypeCurrentRPC = serializeTypeConfigInThisServer;

    //有效消息体
    private transient byte[] body;

    //消息是否要被挂起，现在还用不到这个成员变量，在课程的第二部分内容就会用到了
    //到时候broker就会非常完善了
    private boolean suspended;

    //计时器，用来计算程序处理消息性能的，在我们的代码中并没有用到这个成员变量
    private transient Stopwatch processTimer;

    //受保护的构造方法，防止在该类之外直接实例化RemotingCommand对象
    protected RemotingCommand() {
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：创建请求对象的方法，在第二版本代码中，该方法在NettyCLient测试类中被调用了
     */
    public static RemotingCommand createRequestCommand(int code, CommandCustomHeader customHeader) {
        //创建RemotingCommand对象的时候，请求ID就已经被赋值了
        RemotingCommand cmd = new RemotingCommand();
        //设置请求类型码
        cmd.setCode(code);
        //给请求头赋值
        cmd.customHeader = customHeader;
        //设置请求版本号
        setCmdVersion(cmd);
        return cmd;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：创建带有响应头的响应对象的方法，在我们现在的代码中，该方法并不会被调用
     */
    public static RemotingCommand createResponseCommandWithHeader(int code, CommandCustomHeader customHeader) {
        RemotingCommand cmd = new RemotingCommand();
        cmd.setCode(code);
        cmd.markResponseType();
        cmd.customHeader = customHeader;
        setCmdVersion(cmd);
        return cmd;
    }


    //设置消息版本号的方法
    protected static void setCmdVersion(RemotingCommand cmd) {
        //如果配置版本号已经设置了，那就直接使用该版本号
        if (configVersion >= 0) {
            cmd.setVersion(configVersion);
        } else {
            //如果没有配置，那就从系统环境变量中获得，然后设置到消息中
            String v = System.getProperty(REMOTING_VERSION_KEY);
            if (v != null) {
                //把版本号转换为整数
                int value = Integer.parseInt(v);
                //设置消息版本号
                cmd.setVersion(value);
                //更新配置版本号
                configVersion = value;
            }
        }
    }


    //创建响应码为系统错误的响应对象的方法
    public static RemotingCommand createResponseCommand(Class<? extends CommandCustomHeader> classHeader) {
        return createResponseCommand(RemotingSysResponseCode.SYSTEM_ERROR, "not set any response code", classHeader);
    }

    //创建失败响应对象的方法
    public static RemotingCommand buildErrorResponse(int code, String remark, Class<? extends CommandCustomHeader> classHeader) {
        final RemotingCommand response = RemotingCommand.createResponseCommand(classHeader);
        //设置响应错误码
        response.setCode(code);
        response.setRemark(remark);
        return response;
    }


    public static RemotingCommand buildErrorResponse(int code, String remark) {
        return buildErrorResponse(code, remark, null);
    }



    //创建包含响应头的响应对象的方法
    public static RemotingCommand createResponseCommand(int code, String remark, Class<? extends CommandCustomHeader> classHeader) {
        RemotingCommand cmd = new RemotingCommand();
        cmd.markResponseType();
        cmd.setCode(code);
        cmd.setRemark(remark);
        setCmdVersion(cmd);
        if (classHeader != null) {
            try {
                CommandCustomHeader objectHeader = classHeader.getDeclaredConstructor().newInstance();
                cmd.customHeader = objectHeader;
            } catch (InstantiationException e) {
                return null;
            } catch (IllegalAccessException e) {
                return null;
            } catch (InvocationTargetException e) {
                return null;
            } catch (NoSuchMethodException e) {
                return null;
            }
        }
        return cmd;
    }




    //创建不包含响应头的响应对象的方法
    public static RemotingCommand createResponseCommand(int code, String remark) {
        return createResponseCommand(code, remark, null);
    }

    //将字节数组解码为RemotingCommand对象的方法
    public static RemotingCommand decode(final byte[] array) throws RemotingCommandException {
        ByteBuffer byteBuffer = ByteBuffer.wrap(array);
        return decode(byteBuffer);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：把缓冲区中的数据解码成RemotingCommand对象的方法
     */
    public static RemotingCommand decode(final ByteBuffer byteBuffer) throws RemotingCommandException {
        return decode(Unpooled.wrappedBuffer(byteBuffer));
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：把缓冲区中的数据解码成RemotingCommand对象的方法，在Netty的NettyDecoder解码器中，调用的就是这个方法
     */
    public static RemotingCommand decode(final ByteBuf byteBuffer) throws RemotingCommandException {
        //得到缓冲区中的可读字节的长度
        int length = byteBuffer.readableBytes();
        System.out.println("帧解码器解码后收到的的总字节"+length);
        //读取这部分数据的前4个字节，在编码的时候，缓冲区的最开始4个字节代表消息的总长度
        //但是netty的帧解码器在解码消息的时候，已经把最开始的4个字节读取了，所以到这里缓冲区中的数据就变成了消息头长度+消息头+消息有效数据
        //我特意在代码中添加了几行输出到控制台的信息，大家可以看到客户端发送过来的字节为137，但这里解码后接收到的为133，因为有4个字节已经被读取了
        //这里得到的就是消息头的长度，这么说其实不太准确，因为这4个字节包含了消息头长度和序列化方式
        int oriHeaderLen = byteBuffer.readInt();
        //得到消息头的长度
        int headerLength = getHeaderLength(oriHeaderLen);
        //判断消息头长度是否正确，不正确则抛出异常
        if (headerLength > length - 4) {
            throw new RemotingCommandException("decode error, bad header length: " + headerLength);
        }
        //getProtocolType(oriHeaderLen)得到的就是序列化方式，然后对消息头解码
        RemotingCommand cmd = headerDecode(byteBuffer, headerLength, getProtocolType(oriHeaderLen));
        //计算出有效消息体的长度
        int bodyLength = length - 4 - headerLength;
        byte[] bodyData = null;
        if (bodyLength > 0) {
            bodyData = new byte[bodyLength];
            //从缓冲区中把有效消息的数据写入到字节数组中
            byteBuffer.readBytes(bodyData);
        }
        //把有效消息体交给给当前消息对象
        cmd.body = bodyData;
        return cmd;
    }

    //得到消息头长度的方法
    public static int getHeaderLength(int length) {
        return length & 0xFFFFFF;
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：对消息头解码的方法，这个方法的内容非常简单，我就不添加详细注释了
     */
    private static RemotingCommand headerDecode(ByteBuf byteBuffer, int len, SerializeType type) throws RemotingCommandException {
        //根据序列化类型执行不同的操作
        switch (type) {
            case JSON:
                byte[] headerData = new byte[len];
                byteBuffer.readBytes(headerData);
                RemotingCommand resultJson = RemotingSerializable.decode(headerData, RemotingCommand.class);
                resultJson.setSerializeTypeCurrentRPC(type);
                return resultJson;
            case ROCKETMQ:
                RemotingCommand resultRMQ = RocketMQSerializable.rocketMQProtocolDecode(byteBuffer, len);
                resultRMQ.setSerializeTypeCurrentRPC(type);
                return resultRMQ;
            default:
                break;
        }
        return null;
    }

    //得到序列化协议的方法
    public static SerializeType getProtocolType(int source) {
        return SerializeType.valueOf((byte) ((source >> 24) & 0xFF));
    }

    //创建请求Id的方法
    public static int createNewRequestId() {
        return requestId.getAndIncrement();
    }


    //以下都是一些简单的get、set方法，我就不添加详细注释了


    public static SerializeType getSerializeTypeConfigInThisServer() {
        return serializeTypeConfigInThisServer;
    }


    //标记序列化类型的方法
    public static int markProtocolType(int source, SerializeType type) {
        return (type.getCode() << 24) | (source & 0x00FFFFFF);
    }

    //设置消息为响应类型的方法
    public void markResponseType() {
        int bits = 1 << RPC_TYPE;
        this.flag |= bits;
    }


    public CommandCustomHeader readCustomHeader() {
        return customHeader;
    }


    public void writeCustomHeader(CommandCustomHeader customHeader) {
        this.customHeader = customHeader;
    }



    //解码请求头或响应头的方法
    public CommandCustomHeader decodeCommandCustomHeader(Class<? extends CommandCustomHeader> classHeader) throws RemotingCommandException {
        return decodeCommandCustomHeader(classHeader, false);
    }

    //解码请求头或响应头的方法
    public CommandCustomHeader decodeCommandCustomHeader(Class<? extends CommandCustomHeader> classHeader, boolean isCached) throws RemotingCommandException {
        if (isCached && cachedHeader != null) {
            return cachedHeader;
        }
        cachedHeader = decodeCommandCustomHeaderDirectly(classHeader, true);
        return cachedHeader;
    }




    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：解码请求头或响应头的方法
     */
    public CommandCustomHeader decodeCommandCustomHeaderDirectly(Class<? extends CommandCustomHeader> classHeader, boolean useFastEncode) throws RemotingCommandException {
        CommandCustomHeader objectHeader;
        try {
            //根据请求头类或者响应头型，使用反射获得请求头对象
            objectHeader = classHeader.getDeclaredConstructor().newInstance();
        } catch (InstantiationException e) {
            return null;
        } catch (IllegalAccessException e) {
            return null;
        } catch (InvocationTargetException e) {
            return null;
        } catch (NoSuchMethodException e) {
            return null;
        }
        //这个map中的数据在发送消息过来的时候就会填充完毕，也就是说编码的时候就会对这个成员变量赋值
        if (this.extFields != null) {
            //如果请求头或者响应头实现了快速编码接口
            if (objectHeader instanceof FastCodesHeader && useFastEncode) {
                //那么这里就是用快速解码方式得到请求头或响应头对象
                //这里其实就是直接从extFields中得到对应的value设置到当前解码的CommandCustomHeader对象的成员变量中
                //在第二版本代码中并没有引入任何FastCodesHeader接口的实现类
                ((FastCodesHeader) objectHeader).decode(this.extFields);
                //检验解码后的字段
                objectHeader.checkFields();
                //返回CommandCustomHeader对象
                return objectHeader;
            }
            //程序执行到这里就意味者不能快速解码，那这里就可以先获得当前解码的CommandCustomHeader对象所属类型的所有字段
            //包括该类型父类的字段
            Field[] fields = getClazzFields(classHeader);
            //遍历每一个字段
            for (Field field : fields) {
                //静态字段并不属于任何对象，而是所有对象共享的
                if (!Modifier.isStatic(field.getModifiers())) {
                    //如果不是静态修饰符修饰的字段，这里就得到字段的名称
                    String fieldName = field.getName();
                    //如果字段不是以this开头的
                    if (!fieldName.startsWith("this")) {
                        try {
                            //就根据字段的名称从extFields中获得对应的value，这里得到的value是一个字符串，可能并不是真正和字段对应的类型
                            String value = this.extFields.get(fieldName);
                            //判断字段值是否为null
                            if (null == value) {
                                //如果为null就判断是否允许它为null
                                if (!isFieldNullable(field)) {
                                    //不允许则抛出异常
                                    throw new RemotingCommandException("the custom field <" + fieldName + "> is null");
                                }
                                continue;
                            }
                            //以下就是把从extFields中获得的value转换成字段类型的操作了
                            field.setAccessible(true);
                            //得到字段的类型
                            String type = getCanonicalName(field.getType());
                            Object valueParsed;
                            //如果字段是字符串类型则直接使用
                            if (type.equals(STRING_CANONICAL_NAME)) {
                                valueParsed = value;
                            }
                            //为整数类型则解析为整数类型
                            else if (type.equals(INTEGER_CANONICAL_NAME_1) || type.equals(INTEGER_CANONICAL_NAME_2)) {
                                valueParsed = Integer.parseInt(value);
                            }
                            //为长整数类型则解析为长整数类型
                            else if (type.equals(LONG_CANONICAL_NAME_1) || type.equals(LONG_CANONICAL_NAME_2)) {
                                valueParsed = Long.parseLong(value);
                            }
                            //为布尔类型则解析为布尔类型
                            else if (type.equals(BOOLEAN_CANONICAL_NAME_1) || type.equals(BOOLEAN_CANONICAL_NAME_2)) {
                                valueParsed = Boolean.parseBoolean(value);
                            }
                            //为浮点数类型则解析为浮点数类型
                            else if (type.equals(DOUBLE_CANONICAL_NAME_1) || type.equals(DOUBLE_CANONICAL_NAME_2)) {
                                valueParsed = Double.parseDouble(value);
                            }
                            //这个边界类型暂时还用不上，等到了课程第二部分内容的时候，大家就知道这个类型的作用了
                            else if (type.equals(BOUNDARY_TYPE_CANONICAL_NAME)) {
                                valueParsed = BoundaryType.getType(value);
                            }
                            else {
                                //类型不支持则抛出异常
                                throw new RemotingCommandException("the custom field <" + fieldName + "> type is not supported");
                            }
                            //把类型转换完毕的value设置到字段中
                            field.set(objectHeader, valueParsed);
                        } catch (Throwable e) {
                            log.error("Failed field [{}] decoding", fieldName, e);
                        }
                    }
                }
            }
            objectHeader.checkFields();
        }
        return objectHeader;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：得到指定请求头类型中的所有字段的方法，就是把这个类以及其父类中的字段以反射的方式收集到一个Field数组中，方法的逻辑非常简单，我就不添加详细注释了
     */
    Field[] getClazzFields(Class<? extends CommandCustomHeader> classHeader) {
        Field[] field = CLASS_HASH_MAP.get(classHeader);
        if (field == null) {
            Set<Field> fieldList = new HashSet<>();
            for (Class className = classHeader; className != Object.class; className = className.getSuperclass()) {
                Field[] fields = className.getDeclaredFields();
                fieldList.addAll(Arrays.asList(fields));
            }
            field = fieldList.toArray(new Field[0]);
            synchronized (CLASS_HASH_MAP) {
                CLASS_HASH_MAP.put(classHeader, field);
            }
        }
        return field;
    }


    private boolean isFieldNullable(Field field) {

        if (!NULLABLE_FIELD_CACHE.containsKey(field)) {
            Annotation annotation = field.getAnnotation(CFNotNull.class);
            synchronized (NULLABLE_FIELD_CACHE) {
                NULLABLE_FIELD_CACHE.put(field, annotation == null);
            }
        }
        return NULLABLE_FIELD_CACHE.get(field);
    }


    //得到类的标准名称
    private String getCanonicalName(Class clazz) {
        String name = CANONICAL_NAME_CACHE.get(clazz);
        if (name == null) {
            name = clazz.getCanonicalName();
            synchronized (CANONICAL_NAME_CACHE) {
                CANONICAL_NAME_CACHE.put(clazz, name);
            }
        }
        return name;
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：对当前消息进行编码的方法
     */
    public ByteBuffer encode() {
        //预留4个字节，用来向缓冲区写入消息头长度
        int length = 4;
        //对消息头编码，得到消息头字节数组
        byte[] headerData = this.headerEncode();
        //把消息头长度添加到消息总长度中
        length += headerData.length;
        //如果有效数据不为空，就把有效数据长度添加到消息总长度中
        if (this.body != null) {
            length += body.length;
        }
        //在这类申请缓冲区用来存放编码后的消息，注意，这里申请缓冲区时还增加了4个字节，这4个字节就是用来写入消息总长度的
        ByteBuffer result = ByteBuffer.allocate(4 + length);
        //把消息总长度写入到缓冲区
        result.putInt(length);
        //把消息头长度写入到缓冲区，并且写入序列化类型
        result.putInt(markProtocolType(headerData.length, serializeTypeCurrentRPC));
        //把消息头写入到缓冲区
        result.put(headerData);
        //把消息体写入到缓冲区
        if (this.body != null) {
            result.put(this.body);
        }
        //写变成读
        result.flip();
        return result;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：对消息头编码，得到消息头的字节数组的方法
     */
    private byte[] headerEncode() {
        this.makeCustomHeaderToNet();
        if (SerializeType.ROCKETMQ == serializeTypeCurrentRPC) {
            return RocketMQSerializable.rocketMQProtocolEncode(this);
        } else {
            return RemotingSerializable.encode(this);
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：将请求头或者响应头中的数据存储到extFields中的方法。这个方法使用了反射，把请求头对象或者响应头对象以及其父类对象中的属性都收集到一起然后存放到extFields中了
     * 方法的逻辑非常简单，我就不添加详细注释了
     */
    public void makeCustomHeaderToNet() {
        if (this.customHeader != null) {
            Field[] fields = getClazzFields(customHeader.getClass());
            if (null == this.extFields) {
                this.extFields = new HashMap<>();
            }
            for (Field field : fields) {
                if (!Modifier.isStatic(field.getModifiers())) {
                    String name = field.getName();
                    if (!name.startsWith("this")) {
                        Object value = null;
                        try {
                            field.setAccessible(true);
                            value = field.get(this.customHeader);
                        } catch (Exception e) {
                            log.error("Failed to access field [{}]", name, e);
                        }
                        if (value != null) {
                            this.extFields.put(name, value.toString());
                        }
                    }
                }
            }
        }
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：对请求头或者响应头进行编码的方法
     */
    public void fastEncodeHeader(ByteBuf out) {
        //得到消息的有效数据的字节长度
        int bodySize = this.body != null ? this.body.length : 0;
        //得到缓冲区开始写入数据的索引
        int beginIndex = out.writerIndex();
        //这里的操作就是提前留出8个字节的长度，这个8个字节需要用来写入消息总长度和消息头长度
        out.writeLong(0);
        //定义一个变量，用来接收消息头长度，注意，消息头并不是只包含请求头或响应头的内容
        int headerSize;
        //判断消息序列化类型
        if (SerializeType.ROCKETMQ == serializeTypeCurrentRPC) {
            if (customHeader != null && !(customHeader instanceof FastCodesHeader)) {
                //把消息头中的数据存储到extFields成员变量中
                this.makeCustomHeaderToNet();
            }
            //对当前消息进行编码，返回编码后的消息头字节长度，这里其实就是把除了有效消息体body外的所有数据都写入到缓冲区中了
            //由此可见消息头其实包含很多数据，并不只是包包含请求头和响应头数据
            headerSize = RocketMQSerializable.rocketMQProtocolEncode(this, out);
        } else {
            //如果是json序列化的方式
            this.makeCustomHeaderToNet();
            //那就在这里使用json序列化当前消息对象
            byte[] header = RemotingSerializable.encode(this);
            //这里同样获得了消息头大小
            headerSize = header.length;
            //把消息头写入到缓冲区中
            out.writeBytes(header);
        }
        //设置消息总长度到缓冲区中
        out.setInt(beginIndex, 4 + headerSize + bodySize);
        //在消息总长度后面写入消息头长度，这里也把序列化类型写入到缓冲区中了
        out.setInt(beginIndex + 4, markProtocolType(headerSize, serializeTypeCurrentRPC));
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：对消息头进行编码的方法
     */
    public ByteBuffer encodeHeader() {
        //这里把消息体的长度传进去了
        return encodeHeader(this.body != null ? this.body.length : 0);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2024/12/28
     * @方法描述：对消息头进行编码的方法，方法参数为消息体的长度，因为要申请缓冲区，缓冲区中就是消息长度+消息头长度+消息头+消息体的长度，所以要知道消息体的长度
     */
    public ByteBuffer encodeHeader(final int bodyLength) {
        //预留四个字节，存储消息头长度
        int length = 4;
        byte[] headerData;
        //对消息头编码，得到消息头的字节数组
        headerData = this.headerEncode();
        //把消息头的长度增加到消息总长度中
        length += headerData.length;
        //再把消息有效体的长度增加到消息总长度中
        length += bodyLength;
        //在这类申请缓冲区用来存放编码后的消息，注意，这里申请缓冲区时还增加了4个字节，这4个字节就是用来写入消息总长度的
        //但这里为什么还要把消息有效体再减去呢？这个问题先留着，等后面实现broker部分功能的时候再来看看
        ByteBuffer result = ByteBuffer.allocate(4 + length - bodyLength);
        //把消息总长度写入到缓冲区中
        result.putInt(length);
        //把消息头长度写入到缓冲区中
        result.putInt(markProtocolType(headerData.length, serializeTypeCurrentRPC));
        //把消息头内容写入到缓冲区中
        result.put(headerData);
        ((Buffer) result).flip();
        return result;
    }

    //标记当前消息为单向消息
    public void markOnewayRPC() {
        int bits = 1 << RPC_ONEWAY;
        this.flag |= bits;
    }


    //判断当前消息是否为单向消息
    @JSONField(serialize = false)
    public boolean isOnewayRPC() {
        int bits = 1 << RPC_ONEWAY;
        return (this.flag & bits) == bits;
    }

    //得到请求码或者响应码
    public int getCode() {
        return code;
    }

    //设置请求码或响应码
    public void setCode(int code) {
        this.code = code;
    }


    //得到消息类型
    @JSONField(serialize = false)
    public RemotingCommandType getType() {
        if (this.isResponseType()) {
            return RemotingCommandType.RESPONSE_COMMAND;
        }
        return RemotingCommandType.REQUEST_COMMAND;
    }

    //判断当前消息是否为响应消息
    @JSONField(serialize = false)
    public boolean isResponseType() {
        int bits = 1 << RPC_TYPE;
        return (this.flag & bits) == bits;
    }


    //以下都是一些简单的get、set方法，我就不添加详细注释了



    public LanguageCode getLanguage() {
        return language;
    }


    public void setLanguage(LanguageCode language) {
        this.language = language;
    }


    public int getVersion() {
        return version;
    }


    public void setVersion(int version) {
        this.version = version;
    }


    public int getOpaque() {
        return opaque;
    }


    public void setOpaque(int opaque) {
        this.opaque = opaque;
    }


    public int getFlag() {
        return flag;
    }


    public void setFlag(int flag) {
        this.flag = flag;
    }


    public String getRemark() {
        return remark;
    }


    public void setRemark(String remark) {
        this.remark = remark;
    }


    public byte[] getBody() {
        return body;
    }


    public void setBody(byte[] body) {
        this.body = body;
    }


    @JSONField(serialize = false)
    public boolean isSuspended() {
        return suspended;
    }


    @JSONField(serialize = false)
    public void setSuspended(boolean suspended) {
        this.suspended = suspended;
    }


    public HashMap<String, String> getExtFields() {
        return extFields;
    }


    public void setExtFields(HashMap<String, String> extFields) {
        this.extFields = extFields;
    }


    public void addExtField(String key, String value) {
        if (null == extFields) {
            extFields = new HashMap<>(256);
        }
        extFields.put(key, value);
    }


    public void addExtFieldIfNotExist(String key, String value) {
        extFields.putIfAbsent(key, value);
    }


    @Override
    public String toString() {
        return "RemotingCommand [code=" + code + ", language=" + language + ", version=" + version + ", opaque=" + opaque + ", flag(B)="
                + Integer.toBinaryString(flag) + ", remark=" + remark + ", extFields=" + extFields + ", serializeTypeCurrentRPC="
                + serializeTypeCurrentRPC + "]";
    }


    public SerializeType getSerializeTypeCurrentRPC() {
        return serializeTypeCurrentRPC;
    }


    public void setSerializeTypeCurrentRPC(SerializeType serializeTypeCurrentRPC) {
        this.serializeTypeCurrentRPC = serializeTypeCurrentRPC;
    }


    public Stopwatch getProcessTimer() {
        return processTimer;
    }


    public void setProcessTimer(Stopwatch processTimer) {
        this.processTimer = processTimer;
    }


}
